package com.jwater.core.manager.master.check;

import com.jwater.core.manager.common.DataKeys;
import com.jwater.core.manager.common.model.ReportInfo;
import com.jwater.core.manager.common.model.WorkerInfo;
import com.jwater.core.manager.master.MasterNodeImpl;
import com.jwater.core.net.group.data.DurableDataItem;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

/**
 * @description:
 * @author: jnner
 * @createDate: 2021/3/5
 * @version: 1.0
 */
public class WorkerCheck implements Runnable {
    protected static Logger logger = LoggerFactory.getLogger(WorkerCheck.class);
    private MasterNodeImpl masterNode;
    private Thread thread;
    private Map<String, Integer> counts = new HashMap<>();
    private Map<String, ReportInfo> reportInfoCache = new ConcurrentHashMap<>();

    public WorkerCheck(MasterNodeImpl masterNode) {
        this.masterNode = masterNode;
    }

    public void start() {
        thread = new Thread(this);
        thread.start();
    }

    @Override
    public void run() {
        while (true) {
            try {
                check();
            } catch (Exception e) {
                logger.error(e.toString(), e);
            }
            synchronized (this) {
                try {
                    this.wait(1000);
                } catch (InterruptedException e) {
                    logger.error(e.toString(), e);
                }
            }
        }
    }

    /**
     * 检查多个进程同时存在的情况
     *
     * @param reportInfo
     */
    public void check(ReportInfo reportInfo) {
        if (masterNode.groupNode == null || !masterNode.groupNode.isLeader()) {
            return;
        }

        ReportInfo last = reportInfoCache.get(reportInfo.getKey());
        if (last != null && last.getPid() != reportInfo.getPid()) {
            if (reportInfo.getUptime() > last.getUptime()) {
                masterNode.dispatchRepeatWorkerEvent(last.getKey(), last.getHost());
            } else {
                masterNode.dispatchRepeatWorkerEvent(reportInfo.getKey(), reportInfo.getHost());
            }
        }
        reportInfoCache.put(reportInfo.getKey(), reportInfo);
    }

    public void remove(String key) {
        reportInfoCache.remove(key);
    }

    public void check() {
        if (masterNode.groupNode == null || !masterNode.groupNode.isLeader()) {
            return;
        }

        Map<String, DurableDataItem> runs = masterNode.groupNode.getItems(DataKeys.KEY_RUN);
        for (Map.Entry<String, DurableDataItem> en : runs.entrySet()) {
            DurableDataItem online = masterNode.groupNode.getDataItem(DataKeys.KEY_ONLINE, en.getKey());
            if (online == null) {
                Integer count = counts.get(en.getKey());
                if (count == null) {
                    count = 0;
                }
                count++;
                //连续10次都不在线，则启动程序
                if (count > 10) {
                    String[] items = en.getKey().split(":");
                    logger.info("check reboot worker:" + en.getKey());
                    masterNode.startWorker(items[0], items[1], items[2], items[3], Integer.parseInt(items[4]));
                    // 设置下次重启时间
                    counts.put(en.getKey(), 5);
                } else {
                    counts.put(en.getKey(), count);
                }
            } else {
                counts.remove(en.getKey());
            }
        }
    }
}
